package com.poizon.infrastructure.idem.core.util.impl;

import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.poizon.infrastructure.idem.core.domain.entity.IdemEntity;
import com.poizon.infrastructure.idem.core.domain.value.exception.IdempotentDbStoreException;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

import static com.poizon.infrastructure.idem.core.domain.value.config.IdemConstants.IDEM_RETRY_WAIT_TIME;

/**
 * 工具类/幂等
 */
@Slf4j
public class IdemUtil extends AbstractIdemUtil {
    public IdemUtil(ThreadPoolExecutor idemAsyncSubmitRespExecutor) {
        super(idemAsyncSubmitRespExecutor);
    }

    /**
     * 幂等操作
     * todo= 幂等业务指标监控采集,幂等业务数,成功业务数,失败业务数,并发冲突业务数 DB底层存储异常飞书告警
     */
    @Override
    protected Object executeIdemImpl(String operationKey, Supplier serviceRunnable, Long expireTime, @NotNull TimeUnit timeUnit, @NotBlank String reqType, @NotNull AtomicInteger retryCount) {
        String idemKey = Get.getIdemKey(operationKey);
        String threadName = Get.getThreadName();
        try {
            // 获取幂等对象
            IdemEntity idemEntity = IdemEntity.Get.getIdemDto(idemKey);
            if (idemEntity != null) {
                // 已经存在(不存在并发问题)
                return idemObjHasExist(operationKey, serviceRunnable, expireTime, timeUnit, reqType, retryCount, idemEntity, idemKey);
            } else {
                // 幂等实体不存在
                return idemObjNotExist(operationKey, serviceRunnable, expireTime, timeUnit, reqType, retryCount, idemKey, threadName);
            }
        } catch (IdempotentDbStoreException e) {
            log.error(StrUtil.format(">幂等底层存储异常"), e);
            throw e;
        } catch (Exception e) {
            throw e;
        }
    }

    /**
     * 已经执行过幂等,但是数据没刷(主要体现在数据库),也有可能幂等操作还在执行中,可能执行失败,导致幂等回退,所以要重试
     * 重试包含两部分= 并发插入失败重试 + 幂等建存在,数据不存在重试
     */
    @SneakyThrows
    private Object retry(@NotBlank String operationKey, @NotNull Long expireTime, @NotBlank String reqType, @NotNull Supplier serviceRunnable, @NotNull AtomicInteger retryCount, @NotNull TimeUnit timeUnit) {
        String idemKey = Get.getIdemKey(operationKey);

        log.debug(">幂等key={},并发冲突,休眠重试", idemKey);

        Thread.sleep(IDEM_RETRY_WAIT_TIME);

        log.debug(">休眠结束,准备尝试二次获取幂等key={}的数据", idemKey);

        int retryTimes = retryCount.getAndDecrement();
        if (retryTimes <= 0) {
            String errorMsg = StrUtil.format(">告警: 幂等key={},到达重试次数,尚未获取到幂等响应,可能数据落库失败,请人工手动处理", idemKey);
            log.error(errorMsg);
            throw new IdempotentDbStoreException(errorMsg);
        }

        log.debug(">幂等key={},开始倒数第{}次重试", idemKey, retryTimes);

        return executeIdemImpl(operationKey, serviceRunnable, expireTime, timeUnit, reqType, retryCount);
    }

/**
 * 一开始进来,判断已经获取到幂等对象
 */
    /**
     * 幂等对象已经存在
     */
    private Object idemObjHasExist(@NotBlank String operationKey, @NotNull Supplier serviceRunnable, @NotNull Long expireTime, @NotNull TimeUnit timeUnit, @NotBlank String reqType, @NotNull AtomicInteger retryCount, @NotNull IdemEntity idemEntity, @NotBlank String idemKey) {
        Assert.notNull(idemEntity.getStatus(), "幂等对象状态为空");

        Object resp = idemEntity.getResp();
        Boolean status = idemEntity.getStatus();

        if (status) {
            // resp已落库
            if (log.isDebugEnabled()) {
                log.debug(">幂等key={},重复执行,响应已落库,直接返回,resp={}", idemKey, JSON.toJSONString(resp));
            }

            return idemObjHasExistAndHasResp(resp);
        } else {
            // resp未落库
            log.debug(">幂等key={},重复执行,响应未落库,准备重试,重试次数={}", idemKey, retryCount.get());
            return idemObjHasExistAndNoResp(operationKey, serviceRunnable, expireTime, timeUnit, reqType, retryCount);
        }
    }

    private Object idemObjHasExistAndHasResp(Object resp) {
        return resp;
    }


    private Object idemObjHasExistAndNoResp(@NotBlank String operationKey, @NotNull Supplier serviceRunnable, @NotNull Long expireTime, @NotNull TimeUnit timeUnit, @NotBlank String reqType, @NotNull AtomicInteger retryCount) {
        return retry(operationKey, expireTime, reqType, serviceRunnable, retryCount, timeUnit);
    }

    /**
     * 幂等对象存在且存在并发冲突
     */
    private Object idemObjNotExistAndHasConcurrencyViolation(String operationKey, Supplier serviceRunnable, Long expireTime, @NotNull TimeUnit timeUnit, @NotBlank String reqType, @NotNull AtomicInteger retryCount, @NotBlank String idemKey) {
        if (log.isDebugEnabled()) {
            log.debug(">幂等key={},并发冲突,准备重试,重试次数={}", idemKey, retryCount.get());
        }
        return retry(operationKey, expireTime, reqType, serviceRunnable, retryCount, timeUnit);
    }


/**
 * 一开始进来,没有获取到幂等对象
 */
    /**
     * 幂等对象不存在
     */
    private Object idemObjNotExist(String operationKey, @NotNull Supplier serviceRunnable, @NotNull Long expireTime, @NotNull TimeUnit timeUnit, @NotBlank String reqType, @NotNull AtomicInteger retryCount, @NotBlank String idemKey, String threadName) {
        // 幂等key不存在
        log.debug(">幂等key={},不存在", idemKey);

        // 同步插入唯一键
        IdemEntity emptyRespIdemEntity = IdemEntity.Builder.buildCreateIdemDto(idemKey, reqType);
        Boolean success = emptyRespIdemEntity.setIdemKey(idemKey, emptyRespIdemEntity, expireTime, timeUnit);

        // 并发冲突(设置幂等key)
        if (!success) {
            return idemObjNotExistAndHasConcurrencyViolation(operationKey, serviceRunnable, expireTime, timeUnit, reqType, retryCount, idemKey);
        } else {
            return idemObjNotExistAndNoConcurrencyViolation(serviceRunnable, expireTime, timeUnit, idemKey, threadName, emptyRespIdemEntity);
        }
    }

    /**
     * 幂等对象不存在且无并发冲突
     */
    private Object idemObjNotExistAndNoConcurrencyViolation(Supplier serviceRunnable, Long expireTime, @NotNull TimeUnit timeUnit, @NotBlank String idemKey, String threadName, IdemEntity emptyRespIdemEntity) {
        log.debug(">线程={},幂等键设置成功", threadName);
        log.debug(">开始执行幂等业务逻辑");
        try {
            Object resp = serviceRunnable.get();

            log.debug(">幂等业务执行成功");

            flushIdemResAsync(expireTime, timeUnit, emptyRespIdemEntity, resp, idemKey);
            return resp;
        } catch (Exception e) {
            log.debug(">幂等键={},幂等业务执行失败,准备回退幂等key", idemKey);

            log.debug(">幂等键={},开始回退", idemKey);

            emptyRespIdemEntity.delete(idemKey);
            log.debug(">幂等键={},回退成功", idemKey);
            // 把业务执行异常抛出
            throw e;
        }
    }

    /**
     * 异步刷新幂等响应数据入库
     */
    private void flushIdemResAsync(@NotNull Long expireTime, @NotNull TimeUnit timeUnit, @NotNull IdemEntity emptyRespIdemPo, Object resp, @NotBlank String idemKey) {
        String respJson = JSON.toJSONString(resp);

        log.debug(">准备异步刷新幂等key={}的业务数据到DB={}", idemKey, respJson);

        try {
            // todo= 发送MQ,基于MQ可靠消息,保障消息的可靠插入(消费者数量以及消费线程数,需要根据幂等业务量进行指定)
            idemAsyncSubmitRespExecutor.execute(() -> {
                emptyRespIdemPo.updateIdemResp(resp, expireTime, timeUnit, idemKey, emptyRespIdemPo);
            });
        } catch (Exception e) {
            log.warn(">告警: 线程池刷新幂等key={},业务响应数据={},入库异常,请及时检查线程池状态,自动故障转移至主线程执行,错误信息={}", idemKey, respJson, e.getMessage());

            //自从得物上容器后发现一个严重问题，优雅发布时线程池中的线程还未处理完所有数据，线程池直接被shutdown，
            //导致一旦发版就有部分数据插入失败，这种极少数的极端场景直接用主线程去更新，不借助线程池去操作
            try {
                log.warn(">主线程开始同步插入幂等响应结果,idemKey为={},data={}", idemKey, JSON.toJSONString(emptyRespIdemPo));
                emptyRespIdemPo.updateIdemResp(resp, expireTime, timeUnit, idemKey, emptyRespIdemPo);
            } catch (Exception exception) {
                log.error(">告警: 线程池=>主线程,执行idemKey={}失败,请检查幂等底层存储是否正常,错误信息={},请手动刷新幂等响应结果入库,key={},data={}", idemKey, exception.getMessage(), idemKey, respJson);
                throw new IdempotentDbStoreException(e);
            }
        }
    }

}
